package edu.kit.aifb.ldstream.learner.tasks.mod;

import java.util.Arrays;
import java.util.Set;

import javax.xml.datatype.XMLGregorianCalendar;

import org.openrdf.model.BNode;
import org.openrdf.model.Literal;
import org.openrdf.model.Resource;
import org.openrdf.model.Statement;
import org.openrdf.model.URI;
import org.openrdf.query.BindingSet;
import org.openrdf.query.MalformedQueryException;
import org.openrdf.query.QueryEvaluationException;
import org.openrdf.query.QueryLanguage;
import org.openrdf.query.TupleQuery;
import org.openrdf.query.TupleQueryResult;
import org.openrdf.repository.RepositoryException;
import org.openrdf.repository.RepositoryResult;
import org.semanticweb.yars.nx.namespace.RDF;

import edu.kit.aifb.ldstream.env.Environment;
import edu.kit.aifb.ldstream.stat.util.IDistributionChecker;

public class LearnClasses extends LearnerTask{

	public static final int REFERENCE_YEAR = 1970;
	
	private Set<Type> typeSet;
	private ModStatistics stats;
	private IDistributionChecker dc;

	public LearnClasses(Set<Type> typeSet, ModStatistics stats, IDistributionChecker dc) {
		this.typeSet = typeSet;
		this.dc = dc;
		this.stats = stats;
	}

	@Override
	public void run() {

		startup();

		if(Environment.LOGGING) Environment.log.info("Started to learn the Classes.");
		learnTypes();
		if(Environment.LOGGING) Environment.log.info("Finished learning the Classes.");

		if(Environment.LOGGING) Environment.log.info("Started to learn the Instance distribution.");
		if(Environment.TIME_MEASUREMENTS){
			eventwiseInstanceDistribution();
		} else{
			sequentialInstanceDistribution();
		}
		
		//instanceDistribution();
		if(Environment.LOGGING) Environment.log.info("Finished learning the instance distribution.");

		if(Environment.LOGGING)Environment.log.info("Started to learn the bandwidth distribution.");
		// if time measurments is enabled the "payload" of an event is defined as its outdegree
		if(Environment.TIME_MEASUREMENTS){
			OutdegreeDistribution();
		}else{
			leanBandwidthDistribSequential();
		}
		
		//	learnBandwidthDistrib();
		if(Environment.LOGGING) Environment.log.info("Finished learning the bandwidth distribution.");

		if(Environment.LOGGING)Environment.log.info("Started to learn the In- and Outdegree.");
		inAndOutDegree();
		if(Environment.LOGGING) Environment.log.info("Finished learning In- and Outdegree.");
		
		
		if(Environment.TIME_MEASUREMENTS){
			if(Environment.LOGGING)Environment.log.info("Started to learn the time measurements.");
			timeMeasurements();
			if(Environment.LOGGING) Environment.log.info("Finished learning the time measurements.");
		}
		shutdown();
	}


	private void leanBandwidthDistribSequential(){

		int[] tripleCounts = new int[this.stats.getNumEvents()];
		RepositoryResult<Statement> result = null;
		try {
			result = this.con.getStatements(null, null, null, false);
			Statement s;
			int id;
			try{
				while(result.hasNext()){
					s = result.next();
					//retrieving the context of that triple. The local name is the EventID
					if(s.getContext() != null){
						id = Integer.parseInt(((URI) s.getContext()).getLocalName());
						tripleCounts[id]++;
					}
				}
			} finally{
				result.close();
			}
		} catch (RepositoryException e) {
			e.printStackTrace();
		}
		Integer[] values = new Integer[this.stats.getNumEvents()];
		for(int i = 0; i < values.length; i++){
			values[i] = Integer.valueOf(tripleCounts[i]);
		}
		this.stats.setBandwidthDistribution(this.dc.discrete(values, false));
		System.gc();
	}

	/*
	 * Use leanBandwidthDistribSequential() instead!
	 */
	@Deprecated
	private void learnBandwidthDistrib(){
		String select = "SELECT (COUNT(?pred) AS ?numTriples) ";
		String where = " WHERE {?subj ?pred ?obj}";
		StringBuilder query;
		StringBuilder from; 
		Integer[] bandwidths = new Integer[this.stats.getNumEvents()];

		int tenth = Math.max(this.stats.getNumEvents() / 10, 1);
		for(int i = 0; i < this.stats.getNumEvents(); i++){
			if((i % tenth) == 0){
				System.gc();
				if(Environment.LOGGING) Environment.log.info("Bandwith distribution finished to " + (double) i / ((double) this.stats.getNumEvents())) ; 
			}
			from = new StringBuilder(" FROM <").append(Environment.EVENT_CONTEXT).append(i).append("> ");
			query = new StringBuilder(select).append(from).append(where);
			bandwidths[i] = Integer.valueOf(getCount(query.toString(), "numTriples"));
		}
		this.stats.setBandwidthDistribution(this.dc.discrete(bandwidths, false));
	}

	private void learnTypes(){
		/*
		String query = "SELECT DISTINCT ?type FROM <" + Environment.STREAM_CONTEXT_URI.stringValue() + "> "
						+ "WHERE { ?instance " + RDF.TYPE.toN3() + " ?type FILTER( !isBLANK(?instance)) }";
		 */
		StringBuilder query = new StringBuilder("SELECT DISTINCT ?type WHERE { ?instance ")
		.append(RDF.TYPE.toN3())
		.append(" ?type FILTER( !isBLANK(?instance)) }");
		try {
			TupleQuery tupleQuery = con.prepareTupleQuery(QueryLanguage.SPARQL, query.toString());
			TupleQueryResult result = tupleQuery.evaluate();
			try{
				BindingSet set;
				while(result.hasNext()){
					set = result.next();
					this.typeSet.add(new Type(set.getValue("type").stringValue()));
				}
			}finally{
				result.close();
			}
		} catch (RepositoryException e) {
			e.printStackTrace();
		} catch (MalformedQueryException e) {
			e.printStackTrace();
		} catch (QueryEvaluationException e) {
			e.printStackTrace();
		}
		System.gc();
	}
	
	private void eventwiseInstanceDistribution(){
		Integer[] occurrences;
		for(Type type : this.typeSet){
			occurrences = new Integer[1];
			occurrences[0] = Integer.valueOf(1);
			type.setInstanceOcc(this.dc.discrete(occurrences, false));
		}
		
	}

	private void sequentialInstanceDistribution(){
		RepositoryResult<Statement> result;
		for(Type type : this.typeSet){
			int[] instanceOcc = new int[this.stats.getNumEvents()];
			try {
				result = this.con.getStatements(null, org.openrdf.model.vocabulary.RDF.TYPE, type, false);
				Statement s;
				int eventID;
				try{
					while(result.hasNext()){
						s = result.next();
						//should not be a bnode
						if(s.getSubject() instanceof URI){
							//local name is the event id 
							eventID = Integer.parseInt(((URI) s.getContext()).getLocalName());
							instanceOcc[eventID]++;
						}
					}
				} finally{
					result.close();
				}
			} catch (RepositoryException e) {
				e.printStackTrace();
			}
			Integer[] temp = new Integer[this.stats.getNumEvents()];
			for(int i = 0; i < instanceOcc.length; i++){
				temp[i] = Integer.valueOf(instanceOcc[i]);
			}
			type.setInstanceOcc(this.dc.discrete(temp, false));
			// free memory
			temp = null;
			instanceOcc = null;
			System.gc();
		}
	}

	@Deprecated
	private void instanceDistribution(){

		Integer[] occurrences;
		StringBuilder from;
		StringBuilder where;
		StringBuilder query;
		// Select clause (type and event independent)
		String select = "SELECT (COUNT(?x) AS ?numInstances) ";
		for(Type type : this.typeSet){
			occurrences = new Integer[this.stats.getNumEvents()];
			//Where clause (type dependent)
			where = new StringBuilder(" WHERE {?x ").append(RDF.TYPE.toN3()).append(" <").append(type.stringValue()).append("> FILTER(!isBLANK(?x))}");
			for(int i = 0; i < this.stats.getNumEvents(); i++){
				//build from clause
				from = new StringBuilder(" FROM <").append(Environment.EVENT_CONTEXT).append(i).append("> ");
				// build query
				query = new StringBuilder(select).append(from).append(where);
				occurrences[i] = Integer.valueOf(getCount(query.toString(), "numInstances"));
			}
			type.setInstanceOcc(this.dc.discrete(occurrences, false));
			occurrences = null;
		}

		System.gc();
	}

	private int getCount(String query, String variableName){
		int count = 0;
		try {
			TupleQuery tq= this.con.prepareTupleQuery(QueryLanguage.SPARQL, query);
			TupleQueryResult result = tq.evaluate();
			try{
				if(result.hasNext()){
					count = Integer.parseInt(result.next().getValue(variableName).stringValue());
				}
			}finally{
				result.close();
			}
		} catch (RepositoryException e) {
			e.printStackTrace();
		} catch (MalformedQueryException e) {
			e.printStackTrace();
		} catch (QueryEvaluationException e) {
			e.printStackTrace();
		}
		return count;
	}

	private void inAndOutDegree(){
		RepositoryResult<Statement> result = null;
		Statement s;
		int numInstances = 0;
		int inLinks = 0;
		int outLinks = 0;
		
		for(Type t : this.typeSet){
			try{
				// get all instances
				result = this.con.getStatements(null, org.openrdf.model.vocabulary.RDF.TYPE, t, false);
				result.enableDuplicateFilter();
				while(result.hasNext()){
					s = result.next();
					numInstances++;
					//calc outdegree
					outLinks += countOutLinks(s.getSubject());
					//indegree
					inLinks += countInLinks(s.getSubject());
				}

			} catch (RepositoryException e) {
				e.printStackTrace();
			}finally{
				try {
					if(result != null)
						result.close();
				} catch (RepositoryException e) {
					e.printStackTrace();
				}
			}
		}
		// save results
		if(numInstances > 0){
			this.stats.setAvgIndegree((double) inLinks / (double) numInstances);
			this.stats.setAvgOutdegree((double) outLinks / (double) numInstances );
		}

	}

	private int countInLinks(Resource instance){
		int inLinks = 0;
		RepositoryResult<Statement> result = null;
		try {
			result = this.con.getStatements(null, null, instance, false);
			while(result.hasNext()){
				result.next();
				inLinks++;
			}
		} catch (RepositoryException e) {
			e.printStackTrace();
		} finally{
			try {
				if(result != null){
					result.close();
				}
			} catch (RepositoryException e) {
				e.printStackTrace();
			}
		}
		
		return inLinks;
	}

	private int countOutLinks(Resource instance){
		int outLinks = 0;
		RepositoryResult<Statement> result = null;
		Statement s;
		
		try {
			result = this.con.getStatements(instance, null, null, false);
			while(result.hasNext()){
				s = result.next();
				if(s.getObject() instanceof BNode){
					// get all the links behind the bnode
					outLinks += countOutLinks((BNode) s.getObject());
					//also count the link to the bnode
					outLinks++;
				}else{
					outLinks++;
				}
			}
		} catch (RepositoryException e) {
			e.printStackTrace();
		} finally{
			try {
				if(result != null){
					result.close();
				}
			} catch (RepositoryException e) {
				e.printStackTrace();
			}
		}
		return outLinks;
	}


	private void timeMeasurements(){
		long[] timeStamps = new long[this.stats.getNumInstances()];
		RepositoryResult<Statement> result = null;
		Statement s;
		Literal lit;
		int i = 0;
		try {
			result = this.con.getStatements(null, this.con.getValueFactory().createURI("http://events.event-processing.org/types/endTime"), null, false);
			
			while(result.hasNext() && i < timeStamps.length){
				s = result.next();
				if(s.getObject() instanceof Literal){
					lit = (Literal) s.getObject();
					timeStamps[i++] = getRelativeMillis(lit.calendarValue());
				}
			}
		} catch (RepositoryException e) {
			e.printStackTrace();
		} finally{
			if(result != null){
				try {
					result.close();
				} catch (RepositoryException e) {
					e.printStackTrace();
				}
			}
		}
		
		if(i > 1){
			// no guarantee for correct ordering of the triple results
			Arrays.sort(timeStamps);
			Integer[] differences = compDifferences(timeStamps);
			this.stats.setFrequncyDistribution(this.dc.discrete(differences, false));
			Integer[] jitter = compJitter(differences);
			this.stats.setJitterDistribution(this.dc.discrete(jitter, false));
		}
		
	}
	
	private Integer[] compDifferences(long[] timeStamps){
		// calculate the differences between two following events
		Integer[] differences = new Integer[timeStamps.length - 1];
		for(int i = 0; i < differences.length; i++){
			differences[i] = Integer.valueOf((int)(timeStamps[i + 1] - timeStamps[i]));
		}
		return differences;
	}
	
	private Integer[] compJitter(Integer[] differences){
		// here jitter is the difference of two following frequency differences
		Integer[] jitter = new Integer[differences.length - 1];
		for(int i = 0; i < jitter.length; i++){
			jitter[i] = Integer.valueOf(Math.abs(differences[i + 1].intValue() - differences[i].intValue())); 
		}
		return jitter;
	}
	
	/*
	 * this method is for computing the payload for event-based data 
	 */
	private void OutdegreeDistribution(){
		RepositoryResult<Statement> result = null;
		Statement s;
		Integer[] outdegrees = new Integer[this.stats.getNumInstances()];
		int i = 0;
		for(Type t : this.typeSet){
			try {
				result = this.con.getStatements(null, org.openrdf.model.vocabulary.RDF.TYPE, t, false);
				while(result.hasNext() && i < outdegrees.length){
					s = result.next();
					outdegrees[i++] = Integer.valueOf(countOutLinks(s.getSubject())); 
				}
			} catch (RepositoryException e) {
				e.printStackTrace();
			} finally{
				if(result != null){
					try {
						result.close();
					} catch (RepositoryException e) {
						e.printStackTrace();
					}
				}
			}
		}
		
		this.stats.setBandwidthDistribution(this.dc.discrete(outdegrees, false));
		
	}
	
	private long getRelativeMillis(XMLGregorianCalendar date){
		// convert to millis
		long millis = date.getMillisecond() 
				    + 1000 * (date.getSecond() 
				    + 60 * (date.getMinute()
				    + 60 * (date.getHour() 
				    + 24 * (date.getDay() 
				    + 31 * (date.getMonth() 
				    // normalize with a reference year to avoid overflow		
				    + 12 * (date.getYear() - REFERENCE_YEAR) ) ))));
		return millis;
	}
}
